// NOTE: since /class/ece428/libs/include/event2 is mysteriously unreadable,
// the Makefile expects that libevent2 is installed locally with
// ./configure --prefix=$HOME && make && make install

#include "NodeService.h"
#include <protocol/TBinaryProtocol.h>
#include <server/TNonblockingServer.h>
#include <transport/TServerSocket.h>
#include <transport/TBufferTransports.h>
#include <transport/TSocket.h>
#include <concurrency/ThreadManager.h>
#include <concurrency/PosixThreadFactory.h>

#include <pthread.h>
#include <iostream>
#include <getopt.h>
#include <stdarg.h>
#include <assert.h>
#include <stdlib.h>
#include <math.h>
#include <map>

#include "util.h"
#include "WrappedNode.h"

using namespace ::apache::thrift;
using namespace ::apache::thrift::protocol;
using namespace ::apache::thrift::transport;
using namespace ::apache::thrift::server;
using namespace ::apache::thrift::concurrency;

using boost::shared_ptr;
using namespace  ::mp2;
using namespace std;

pthread_t stabilize_thread;
pthread_t finger_thread;
pthread_mutex_t files_lock;

Node* myNode = NULL;
Node* predecessor = NULL;

int stabilizeInterval;
int fixInterval;

std::vector<Node> finger;
std::map<int32_t,File> files;

class NodeServiceHandler : virtual public NodeServiceIf {
    public:
        NodeServiceHandler() {}

        // findSuccessor: per pseudocode in Chord paper, find the node that
        // immediately follows or lies at the given hash on the ring
        void findSuccessor(Node& _return, const int32_t hash) {
            // Find predecessor
            Node pred;
            findPredecessor(pred, hash);

            // return successor of the predecessor
            _return = WrappedNode(pred).getSuccessor().N;
        }

        // findPredecessor: per pseudocode in Chord paper, find the node that
        // lies immediately before the given hash on the ring
        void findPredecessor(Node& _return, const int32_t hash) {
            // fast path if we are the predecessor
            if (rangecheck_inclusive(myNode->hash, finger[0].hash, hash)) {
                _return = *myNode;
                return;
            }

            Node n;
            closestPrecedingFinger(n, hash);

            WrappedNode cur(n);

            for (; !cur.precedes(hash); cur = cur.closestPrecedingFinger(hash));

            _return = cur.N;
        }

        // closestPrecedingFinger: per pseudocode in Chord paper, find the
        // closest finger table entry counter-clockwise from the given hash on
        // the ring
        void closestPrecedingFinger(Node& _return, const int32_t hash) {
            // go backward through the table returning if we find an entry that precedes hash
            for(vector<Node>::reverse_iterator rit = finger.rbegin(); rit != finger.rend(); ++rit){
                Node cur = *rit;
                if (rangecheck_exclusive(myNode->hash, hash, cur.hash)) {
                    _return = cur;
                    return;
                }
            }

            if (rangecheck_exclusive(myNode->hash, hash, finger[0].hash)) {
                _return = finger[0];
            } else {
                _return = *myNode;
            }
        }

        // getSuccessor: return this node's successor
        void getSuccessor(Node& _return) {
            _return = finger[0];
        }

        // getPredecessor: return this node's predecessor
        void getPredecessor(Node& _return) {
            _return = *predecessor;
        }

        // getNode: return this node's Node struct
        void getNode(Node& _return) {
            _return = *myNode;
        }

        // notify: tell this node that another node thinks it is this node's
        // predecessor
        void notify(const Node& n) {
            if (predecessor == NULL || rangecheck_exclusive(predecessor->hash, myNode->hash, n.hash)) {
                cout << "node= " << myNode->id << ": updated predecessor= " << n.id << "\n";
                predecessor = new Node(n);
            }
            pthread_mutex_lock(&files_lock);
            for (map<int32_t, File>::iterator it = files.begin(); it != files.end(); it++) {
                if (!rangecheck_inclusive(predecessor->hash, myNode->hash, it->first)) {
                    WrappedNode(*predecessor).putFile(it->second.name, it->second.data);
                    files.erase(it);
                    cout << "node= " << myNode->id << ": deleted file: k= " << it->first << "\n";
                }
            }
            pthread_mutex_unlock(&files_lock);
        }

        // addFile: add a file to the Chord network with the given filename and
        // contents, first finding the appropriate node to store it, then
        // placing it into the map
        void addFile(std::string& _return, const std::string& filename, const std::string& data) {
            int hash = hashValue(filename.c_str(), myNode->m);

            if (rangecheck_inclusive(predecessor->hash, myNode->hash, hash)) {
                putFile(_return, filename, data);
                _return = get_ADD_FILE_result_as_string(filename.c_str(), hash, myNode->id);
                return;
            }

            Node n;
            findSuccessor(n, hash);

            if (n.hash == myNode->hash) {
                putFile(_return, filename, data);
                _return = get_ADD_FILE_result_as_string(filename.c_str(), hash, myNode->id);
                return;
            }

            _return = WrappedNode(n).addFile(filename, data);
        }

        // putFile: place a file into this node's map: used by addFile to
        // insert files and by notify to move files around without invoking the
        // findSuccessor algorithm on an unstable network
        void putFile(std::string& _return, const std::string& filename, const std::string& data) {
            int hash = hashValue(filename.c_str(), myNode->m);
            File f;
            f.name = filename;
            f.data = data;
            pthread_mutex_lock(&files_lock);
            files[hash] = f;
            pthread_mutex_unlock(&files_lock);
            cout << "node= " << myNode->id << ": added file: k= " << hash << "\n";
        }

        // delFile: delete a file from the Chord network: find the node it
        // should be at, then delete it from the map
        void delFile(std::string& _return, const std::string& filename) {
            int hash = hashValue(filename.c_str(), myNode->m);

            int deleted = 0;
            pthread_mutex_lock(&files_lock);
            if (files.count(hash)) {
                files.erase(hash);
                cout << "node= " << myNode->id << ": deleted file: k= " << hash << "\n";
                deleted = 1;
            }
            pthread_mutex_unlock(&files_lock);

            if (rangecheck_inclusive(predecessor->hash, myNode->hash, hash)) {
                _return = get_DEL_FILE_result_as_string(filename.c_str(), hash, deleted, myNode->id);
                cout << "node= " << myNode->id << ": no such file k= " << hash << " to delete\n";
                return;
            }

            Node n;
            findSuccessor(n, hash);

            if (n.hash == myNode->hash) {
                _return = get_DEL_FILE_result_as_string(filename.c_str(), hash, deleted, myNode->id);
                return;
            }

            _return = WrappedNode(n).delFile(filename);
        }

        // getFile: retrieve a file's contents from the Chord network: locate
        // the node the file should be at, then ask that node for the file's
        // contents
        void getFile(std::string& _return, const std::string& filename) {
            int hash = hashValue(filename.c_str(), myNode->m);

            pthread_mutex_lock(&files_lock);
            if (files.count(hash)) {
                _return = get_GET_FILE_result_as_string(filename.c_str(), hash, true, myNode->id, files[hash].data.c_str());
                pthread_mutex_unlock(&files_lock);
                cout << "node= " << myNode->id << ": served file: k= " << hash << "\n";
                return;
            }
            pthread_mutex_unlock(&files_lock);

            Node n;
            findSuccessor(n, hash);

            if (n.hash == myNode->hash) {
                _return = get_GET_FILE_result_as_string(filename.c_str(), hash, false, myNode->id, NULL);
                cout << "node= " << myNode->id << ": no such file k= " << hash << "to serve\n";
                return;
            }

            _return = WrappedNode(n).getFile(filename);
        }

        // getTable: retrieve the finger table and the file map of the
        // specified node, by ID as given to ADD_NODE: hash the ID, locate its
        // node, and ask it for its table and map
        void getTable(std::string& _return, const int32_t nodeID) {
            if (nodeID == myNode->id) {
                pthread_mutex_lock(&files_lock);
                _return = get_GET_TABLE_result_as_string(finger, myNode->m, myNode->hash, 0, files);
                pthread_mutex_unlock(&files_lock);
                return;
            }

            Node n;
            findSuccessor(n, nodeID);
            if (n.id == nodeID) _return = WrappedNode(n).getTable(nodeID);
        }

};

shared_ptr<NodeServiceHandler> handler;

// stabilize: move the network towards a more-correct state, when called from
// stabilize_thread_main: check our successor's predecessor, to see if it is
// aware of a better successor for us than itself, then notify our successor
// that we think we are its predecessor.
void stabilize() {
    Node x = WrappedNode(finger[0]).getPredecessor().N;
    if (rangecheck_exclusive(myNode->hash, finger[0].hash, x.hash)) {
        finger[0] = x;
    }
    WrappedNode(finger[0]).notify(*myNode);
}

// join: perform initialization on this node: wire up initial successor,
// predecessor, finger tables, file map, and file map mutex
void join(int m, int id, int port, int introducerPort) {
    myNode = new Node();
    Node successor;

    // Populate myNode structure
    myNode->m    = m;
    myNode->id   = id;
    myNode->port = port;
    myNode->introPort = introducerPort;
    myNode->hash = id;

    if (port != introducerPort) {
        // this is not the first node
        WrappedNode introducer(introducerPort);
        successor = introducer.findSuccessor(myNode->hash).N;
        predecessor = new Node(WrappedNode(successor).getPredecessor().N);

        if (successor.hash == myNode->hash) {
            cout << "failing: duplicate node ID\n";
            exit(1);
        }
    } else {
        // this is the first node
        predecessor = myNode;
        successor = *myNode;
    }

    cout << "node= " << myNode->id << ": initial successor= " << successor.id << "\n";

    finger = std::vector<Node>(m, successor);
    files = std::map<int32_t,File>();

    pthread_mutex_init(&files_lock, NULL);
}

// fix_fingers: pick one finger table entry at random and use the findSuccessor
// algorithm to find out what it should be; called periodically from
// finger_thread_main
void fix_fingers() {
    int i = rand() % myNode->m;
    int old = finger[i].id;
    handler->findSuccessor(finger[i], (int)(myNode->hash+pow(2,i)) % (int)pow(2, myNode->m));
    if (old != finger[i].id) {
        cout << "node= " << myNode->id << ": updated finger entry: i= " << i << ", pointer= " << finger[i].id << "\n";
    }
}

// stabilize_thread_main: repeatedly call stabilize every stabilizeInterval
// seconds
void* stabilize_thread_main(void*) {
    for (;;) {
        sleep(stabilizeInterval);
        stabilize();
    }
    return NULL;
}

// fix_fingers_thread_main: repeatedly call fix_fingers every
// fixInterval seconds
void* finger_thread_main(void*) {
    for (;;) {
        sleep(fixInterval);
        fix_fingers();
    }
    return NULL;
}

// main: parse arguments, initialize node, and start Thrift server
int main(int argc, char **argv) {
    int opt;
    int long_index;

    int m = -1;
    int id = -1;
    int port = -1;
    int introducerPort = -1;
    stabilizeInterval = 2;
    fixInterval = 5;
    int seed = -1;
    const char *logconffile = NULL;

    struct option long_options[] = {
        {"m", required_argument, 0, 1000},
        {"id", required_argument, 0, 1001},
        {"port", required_argument, 0, 1002},
        {"introducerPort", required_argument, 0, 1003},
        {"logConf", required_argument, 0, 1004},
		{"stabilizeInterval", required_argument, 0, 1005},
        {"fixInterval", required_argument, 0, 1006},
        {"seed", required_argument, 0, 1007},
        {0, 0, 0, 0},
    };

    // Get all possible arguments
    while ((opt = getopt_long(argc, argv, "", long_options, &long_index)) != -1)
    {
        switch (opt) {
            case 0:
                if (long_options[long_index].flag != 0) {
                    break;
                }
                printf("option %s ", long_options[long_index].name);
                if (optarg) {
                    printf("with arg %s\n", optarg);
                }
                printf("\n");
                break;

            case 1000:
                m = strtol(optarg, NULL, 10);
                //cout << m << "\n";
                //assert((m >= 3) && (m <= 10));
                break;

            case 1001:
                id = strtol(optarg, NULL, 10);
                //cout << id << "\n";
                //assert(id >= 0);
                break;

            case 1002:
                port = strtol(optarg, NULL, 10);
                //cout << port << "\n";
                //assert(port > 0);
                break;

            case 1003:
                introducerPort = strtol(optarg, NULL, 10);
                //cout << introducerPort << "\n";
                //assert(introducerPort > 0);
                break;

            case 1004:
                logconffile = optarg;
                break;

            case 1005:
                stabilizeInterval = strtol(optarg, NULL, 10);
                //assert(stabilizeInterval > 0);
                break;

            case 1006:
                fixInterval = strtol(optarg, NULL, 10);
                //assert(fixInterval > 0);
                break;

            case 1007:
                seed = strtol(optarg, NULL, 10);
                break;

            default:
                exit(1);
        }

    }

    join(m, id, port, introducerPort);

    // start the stabilization and finger-fixing threads
    if (pthread_create(&stabilize_thread, NULL, &stabilize_thread_main, NULL) != 0) {
        fprintf(stderr, "Error in pthread_create\n");
        exit(1);
    }
    if (pthread_create(&finger_thread, NULL, &finger_thread_main, NULL) != 0) {
        fprintf(stderr, "Error in pthread_create\n");
        exit(1);
    }

    handler = shared_ptr<NodeServiceHandler>(new NodeServiceHandler());
    shared_ptr<TProcessor> processor(new NodeServiceProcessor(handler));
    shared_ptr<TProtocolFactory> protocolFactory(new TBinaryProtocolFactory());

    shared_ptr<ThreadManager> threadManager = ThreadManager::newSimpleThreadManager(15);
    shared_ptr<PosixThreadFactory> threadFactory(new PosixThreadFactory());
    threadManager->threadFactory(threadFactory);
    threadManager->start();

    TNonblockingServer server(processor, protocolFactory, port, threadManager);
    server.serve();

    return 0;
}
